package com.hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseDao {

	private static Configuration conf = HBaseConfiguration.create();
	static {
		conf.set("hbase.rootdir", "hdfs://192.168.1.25:9000/hbase");
		// 设置Zookeeper,直接设置IP地址
		conf.set("hbase.zookeeper.quorum", "192.168.1.25");
	}

	// 创建表
	public static void createTable(String tablename, String... columnFamilys) throws Exception {
		Connection connection = ConnectionFactory.createConnection(conf);
		Admin admin = connection.getAdmin();

		TableName tableNameObj = TableName.valueOf(tablename);

		if (admin.tableExists(tableNameObj)) {
			System.out.println("表已经存在！");
			//			System.exit(0);
		} else {
			HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tablename));
			for (String columnFamily : columnFamilys) {
				tableDesc.addFamily(new HColumnDescriptor(columnFamily));
			}
			admin.createTable(tableDesc);
			System.out.println("表创建成功！");
		}
		admin.close();
		connection.close();
	}

	// 删除表
	public static void deleteTable(String tableName) {
		try {
			Connection connection = ConnectionFactory.createConnection(conf);
			Admin admin = connection.getAdmin();
			TableName table = TableName.valueOf(tableName);
			admin.disableTable(table);
			admin.deleteTable(table);
			System.out.println("delete table " + tableName + " ok.");
			admin.close();
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	// 插入一行记录
	public static void addRecord(String tableName, String rowKey, String family, String qualifier, String value) {
		try {
			Connection connection = ConnectionFactory.createConnection(conf);
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
			table.put(put);
			table.close();
			connection.close();
			System.out.println("insert recored " + rowKey + " to table " + tableName + " ok.");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 删除全部数据
	 * @Title: deleteRows 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @return void    返回类型 
	 * @throws
	 */
	public static void deleteRows(String tableName, String rowKey, String family, String qualifier) {
		try {
			Connection connection = ConnectionFactory.createConnection(conf);
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(rowKey.getBytes());
			delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			table.delete(delete);
			System.out.println("删除行成功!");
			table.close();
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 删除最新一条数据
	 * @Title: deleteRow 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @return void    返回类型 
	 * @throws
	 */
	public static void deleteRow(String tableName, String rowKey, String family, String qualifier) {
		try {
			Connection connection = ConnectionFactory.createConnection(conf);
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(rowKey.getBytes());
			delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			table.delete(delete);
			System.out.println("删除行成功!");
			table.close();
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 查询最新一条数据
	 * @Title: getRow 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @return void    返回类型 
	 * @throws
	 */
	public static void getRow(String tableName, String rowKey, String family, String qualifier) {
		try {
			Connection connection = ConnectionFactory.createConnection(conf);
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(rowKey.getBytes());
			get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			Result result = table.get(get);
			byte[] b = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			System.out.println(tableName + ":" + rowKey + ":" + family + ":" + qualifier + ":"
					+ new String(b == null ? new byte[] {} : b));
			table.close();
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void getRows(String tableName, String rowKey, String family, String qualifier) {
		try {
			Connection connection = ConnectionFactory.createConnection(conf);
			Table table = connection.getTable(TableName.valueOf(tableName));
			ResultScanner resultScanner = table.getScanner(Bytes.toBytes(family));
			for (Result result : resultScanner) {
				byte[] b = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
				System.out.println(tableName + ":" + rowKey + ":" + family + ":" + qualifier + ":"
						+ new String(b == null ? new byte[] {} : b));
			}
			table.close();
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) throws Exception {
		HbaseDao.createTable("testTb", "info");
		HbaseDao.addRecord("testTb", "001", "info", "name", "zhangsan");
		HbaseDao.addRecord("testTb", "001", "info", "age", "20");
		HbaseDao.addRecord("testTb", "001", "info", "name", "zhangsan2");
		HbaseDao.addRecord("testTb", "001", "info", "age", "202");

//		HbaseDao.getRow("testTb", "001", "info", "age");
//		HbaseDao.getRow("testTb", "001", "info", "name");
		HbaseDao.getRows("testTb", "001", "info", "age");
		HbaseDao.getRows("testTb", "001", "info", "name");

//		HbaseDao.deleteRow("testTb", "001", "info", "age");
//		HbaseDao.deleteRow("testTb", "001", "info", "name");
//
//		HbaseDao.getRow("testTb", "001", "info", "age");
//		HbaseDao.getRow("testTb", "001", "info", "name");
		HbaseDao.deleteTable("testTb");
	}
}